package com.sjk.flink.functions;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.internal.JsonContext;
import com.sjk.flink.exeception.BaseException;
import com.sjk.flink.exeception.ErrorCode;
import com.sjk.flink.json.KeyConverter;
import com.sjk.flink.pojo.*;
import com.sjk.flink.relevance.RelevanceFactory;
import com.sjk.flink.util.ExecptionUtil;
import com.sjk.flink.util.JsonSchemaUtil;
import com.sjk.flink.util.JzValueUtils;
import com.sjk.flink.util.MessyCodeCheck;
import com.sjk.flink.util.ObjectMapperUtil;
import networknt.schema.ValidationMessage;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 未过期的数据
 *
 * @author SAMA
 */
public class JsonValidProcessFunction extends ProcessFunction<OutPutPojo, OutPutPojo> {
    private static final Logger log = LoggerFactory.getLogger(JsonValidProcessFunction.class);

    private final OutputTag<OutPutPojo> validJsonTag;
    private final OutputTag<OutPutPojo> invalidJsonTag;
    private final OutputTag<OutPutPojo> commonInvalidJsonTag;

    private transient ObjectMapper objectMapper;

    private transient ParameterTool parameterTool;

    @Override
    public void open(Configuration parameters) {
        objectMapper = ObjectMapperUtil.getObjectMapper();
        parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
    }

    private final Map<String, String> SCHEMA_CONFIGS;

    private final Map<String, String> JZZL_CONFIGS;

    private boolean zlfxFlg;

    public JsonValidProcessFunction(boolean zlfxFlg, Map<String, String> SCHEMA_CONFIGS,
            Map<String, String> JZZL_CONFIGS, OutputTag<OutPutPojo> validJsonTag, OutputTag<OutPutPojo> invalidJsonTag,
            OutputTag<OutPutPojo> commonInvalidJsonTag) {
        this.validJsonTag = validJsonTag;
        this.invalidJsonTag = invalidJsonTag;
        this.SCHEMA_CONFIGS = SCHEMA_CONFIGS;
        this.JZZL_CONFIGS = JZZL_CONFIGS;
        this.zlfxFlg = zlfxFlg;
        this.commonInvalidJsonTag = commonInvalidJsonTag;
    }

    @Override
    public void processElement(OutPutPojo outPutPojo, ProcessFunction<OutPutPojo, OutPutPojo>.Context ctx,
            Collector<OutPutPojo> collector) throws Exception {
        log.info("......开始质量分析......");
        try {
            String json = outPutPojo.getJson();
            String zxdm = parameterTool.get("zxdm");
            log.info(zxdm + "       " + json);
            // 不做质量分析
            if (!zlfxFlg) {
                ctx.output(validJsonTag, outPutPojo);
            } else {
                // 数据校准开始
                JsonChecker odsJsonChecker = JsonSchemaUtil.getPatternIdByKey(SCHEMA_CONFIGS, "ODS", json);
                if (!odsJsonChecker.isSuccess()) {
                    String jsonError = objectMapper.writeValueAsString(odsJsonChecker);
                    outPutPojo.setJson(objectMapper.writeValueAsString(odsJsonChecker));
                    log.info("校准失败出力            " + jsonError);
                    // 检查失败
                    ctx.output(invalidJsonTag, outPutPojo);
                } else {
                    // 数据校准&&质量分析
                    log.info("质量分析开始");
                    processRecord(zxdm, ctx, outPutPojo);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            String errorCode = e.getClass().getName();
            if (e instanceof BaseException) {
                errorCode = ((BaseException) e).getCode();
            }
            log.error("系统错误:" + ExecptionUtil.getStackTraceMsg(e));
        }
    }

    /**
     * 校准&&质量检测
     *
     * @param ctx
     * @param outPutPojo
     * @throws Exception
     */
    private void processRecord(String zxdm, ProcessFunction<OutPutPojo, OutPutPojo>.Context ctx, OutPutPojo outPutPojo)
            throws Exception {
        String srcTopic = outPutPojo.getTopic();
        String tableName = outPutPojo.getTableName();
        String arrayJsonStrs = outPutPojo.getJson();
        int clayer_partition = outPutPojo.getClayer_partition();
        long clayer_offset = outPutPojo.getClayer_offset();
        Long clayer_ts = outPutPojo.getClayer_ts();

        JsonNode readTree = objectMapper.readTree(arrayJsonStrs);
        JsonNode recordDataNode = readTree.get("record");
        String recordJson = recordDataNode.asText();
        // 2、校准值替换
        JzPojo jzpojo = JzValueUtils.jsonhander(JZZL_CONFIGS, tableName, recordJson);
        String jzJsonStr = jzpojo.getJson();
        // 校准标志 0 未校准 1 校准成功 2 校准失败
        String jzbz = "0";
        // 校对信息推送
        if (!jzpojo.isSkip()) {
            if (jzpojo.isSuccess()) {
                // 校准成功
                jzbz = "1";
            } else {
                // 城市分区
                int city_partition_id = 0;
                // 城市偏移量
                long city_offset_id = 0;
                // 时间戳
                String city_ts = "";
                JsonNode sscsdmJsonNode = readTree.get("sscsdm");
                // 城市代码
                String topicsscsdm = sscsdmJsonNode.asText();
                // 抽取时间
                // JsonNode extract_timeJsonNode = readTree.get("extract_time");
                // String extract_time = extract_timeJsonNode.asText();
                // 城市分区
                JsonNode city_partition_idJsonNode = readTree.get("city_partition_id");
                if (city_partition_idJsonNode != null) {
                    city_partition_id = new Integer(city_partition_idJsonNode.asText());
                }
                // 城市偏移量
                JsonNode city_offset_idJsonNode = readTree.get("city_offset_id");
                if (city_offset_idJsonNode != null) {
                    city_offset_id = new Long(city_offset_idJsonNode.asText());
                }
                // 时间戳
                JsonNode city_tsJsonNode = readTree.get("city_ts");
                if (city_tsJsonNode != null) {
                    city_ts = city_tsJsonNode.asText();
                }

                // 校准失败
                jzbz = "2";
                List<JzField> fields = jzpojo.getFields();
                if (fields.size() > 0) {
                    for (JzField jzField : fields) {
                        // 把错误信息发送到ECT
                        outPutPojo.setJson(analysisExecption(srcTopic, topicsscsdm, jzpojo.getJson(),
                                jzField.getErrorCode(), jzField.getErrorMesasge(), city_partition_id, city_offset_id,
                                city_ts, clayer_partition, clayer_offset, clayer_ts));
                        //log.info("校准失败");
                        // 解析 JSON 出错，发送到无效数据的侧输出
                        ctx.output(commonInvalidJsonTag, outPutPojo);
                    }
                } else {
                    // 把错误信息发送到ECT
                    outPutPojo.setJson(analysisExecption(srcTopic, topicsscsdm, jzpojo.getJson(), jzpojo.getErrorCode(),
                            jzpojo.getErrorMesasge(), city_partition_id, city_offset_id, city_ts, clayer_partition,
                            clayer_offset, clayer_ts));
                    // 解析 JSON 出错，发送到无效数据的侧输出
                    //log.info("解析失败");
                    ctx.output(commonInvalidJsonTag, outPutPojo);
                }
            }
        }

        JsonNode jcsjzjJsonNode = recordDataNode.get("bsm");
        String jcsjzj = null;
        if (jcsjzjJsonNode != null) {
            jcsjzj = jcsjzjJsonNode.asText();
        }else{
            throw new BaseException("bsm为空>>  "+arrayJsonStrs);
        }

        // 城市分区
        int city_partition_id = 0;
        // 城市偏移量
        long city_offset_id = 0;
        // 时间戳
        String city_ts = "";
        JsonNode sscsdmJsonNode = readTree.get("sscsdm");
        // 城市代码
        String topicsscsdm = sscsdmJsonNode.asText();
        // 抽取时间
        JsonNode extract_timeJsonNode = readTree.get("extract_time");
        String extract_time = extract_timeJsonNode.asText();
        // 城市分区
        JsonNode city_partition_idJsonNode = readTree.get("city_partition_id");
        if (city_partition_idJsonNode != null) {
            city_partition_id = new Integer(city_partition_idJsonNode.asText());
        }
        // 城市偏移量
        JsonNode city_offset_idJsonNode = readTree.get("city_offset_id");
        if (city_offset_idJsonNode != null) {
            city_offset_id = new Long(city_offset_idJsonNode.asText());
        }
        // 时间戳
        JsonNode city_tsJsonNode = readTree.get("city_ts");
        if (city_tsJsonNode != null) {
            city_ts = city_tsJsonNode.asText();
        }

        log.info("开始做schema检查 ... ...");
        // 检查是否
        JsonChecker srcDataJsonChecker = JsonSchemaUtil.getPatternIdByKey(SCHEMA_CONFIGS, srcTopic, jzJsonStr);
        if (!srcDataJsonChecker.isSuccess()) {
            // 获得校对质检文件
            JsonContext jzzlDocument = (JsonContext) JsonPath.parse(JZZL_CONFIGS.get(tableName));
            // 需要校准的文件
            JsonContext srcDocument = (JsonContext) JsonPath.parse(jzJsonStr);

            Set<ValidationMessage> validates = srcDataJsonChecker.getValidate();
            // 获得所有的字段分级
            boolean stdout = true;
            List<String> zdfjs = getZdfjs(jzzlDocument, validates);
            if (zdfjs.isEmpty() || zdfjs.contains("A")) {
                // 写问题
                stdout = false;
            }

            // 字段错误处理
            Iterator<ValidationMessage> iterator = validates.iterator();
            while (iterator.hasNext()) {
                ValidationMessage next = iterator.next();
                outPutPojo.setJson(analysis(jzbz, zxdm, jcsjzj, topicsscsdm, extract_time, srcTopic,
                        srcDataJsonChecker.getExtract_time(), srcDocument, jzzlDocument, next, city_partition_id,
                        city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
                //log.info("QST >>>>>>>>>>>>>>>>>>>>>>>>>>>>" + outPutPojo.getJson());
                // 验证失败
                ctx.output(invalidJsonTag, outPutPojo);

            }
            // 全B类数据同时输出到标准库
            if (stdout) {
                outPutPojo.setJson(packageJson(topicsscsdm, extract_time, srcDataJsonChecker.getData(),
                        city_partition_id, city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
                //log.info("STD >>>>>>>>>>>>>>>>>>>>>>>>>>>>" + outPutPojo.getJson());
                //判断是否有关联性校验
                if(RelevanceFactory.getInstance().relevance(srcTopic)){
                    //做相关性检查
                    boolean check = RelevanceFactory.getInstance().check(srcTopic,srcDataJsonChecker.getData());
                    if(check){
                        // 验证成功
                        ctx.output(validJsonTag, outPutPojo);
                    }else{
                        //相关性检查失败
                        //相关性检查的错误信息
                        outPutPojo.setJson(analysisRelevance(jzbz, zxdm, jcsjzj, topicsscsdm, extract_time, srcTopic,
                                srcDataJsonChecker.getExtract_time(), city_partition_id,
                                city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
                        ctx.output(invalidJsonTag, outPutPojo);
                    }
                }else{
                    ctx.output(validJsonTag, outPutPojo);
                }
            }
        } else {
            outPutPojo.setJson(packageJson(topicsscsdm, extract_time, srcDataJsonChecker.getData(), city_partition_id,
                    city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
            //log.info("STD >>>>>>>>>>>>>>>>>>>>>>>>>>>>" + outPutPojo.getJson());
            // 验证成功
            ctx.output(validJsonTag, outPutPojo);
            //判断是否有关联性校验
            if(RelevanceFactory.getInstance().relevance(srcTopic)){
                //做相关性检查
                boolean check = RelevanceFactory.getInstance().check(srcTopic,srcDataJsonChecker.getData());
                if(check){
                    // 验证成功
                    ctx.output(validJsonTag, outPutPojo);
                }else{
                    //相关性检查失败
                    //相关性检查的错误信息
                    outPutPojo.setJson(analysisRelevance(jzbz, zxdm, jcsjzj, topicsscsdm, extract_time, srcTopic,
                            srcDataJsonChecker.getExtract_time(), city_partition_id,
                            city_offset_id, city_ts, clayer_partition, clayer_offset, clayer_ts));
                    ctx.output(invalidJsonTag, outPutPojo);
                }
            }else{
                ctx.output(validJsonTag, outPutPojo);
            }
        }
    }

    /**
     * private String packageJson(String scrData) throws JsonProcessingException {
     * ObjectNode readTree = (ObjectNode) objectMapper.readTree(scrData);
     * readTree.put("operate_time", JsonSchemaUtil.getNowTime("yyyy-MM-dd
     * HH:mm:ss"));
     * return objectMapper.writeValueAsString(readTree);
     * }
     */

    private String packageJson(String sscddm, String extract_time, String scrData, int city_partition_id,
            long city_offset_id, String city_ts, int clayer_partition, long clayer_offset, Long clayer_ts)
            throws JsonProcessingException {
        ObjectNode readTree = (ObjectNode) objectMapper.readTree(scrData);
        readTree.put("sscsdm", sscddm);
        readTree.put("extract_time", extract_time);
        readTree.put("operate_time", JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        readTree.put("city_partition_id", city_partition_id);
        readTree.put("city_offset_id", city_offset_id);
        readTree.put("city_ts", city_ts);
        readTree.put("clayer_partition", clayer_partition);
        readTree.put("clayer_offset", clayer_offset);
        if (clayer_ts == null) {
            readTree.put("clayer_ts", "");
        } else {
            readTree.put("clayer_ts", formatDate(clayer_ts));
        }
        return objectMapper.writeValueAsString(readTree);
    }

    /**
     * 相关性检查信息
     *
     * @param jcsjzj
     * @param topicsscsdm
     * @param extractTime
     * @param topic
     * @param zjsj
     * @return
     */
    private String analysisRelevance(String jzbz, String zxdm, String jcsjzj, String topicsscsdm, String extractTime,
                            String topic, String zjsj, int city_partition_id, long city_offset_id, String city_ts,
                            int clayer_partition, long clayer_offset, Long clayer_ts) throws JsonProcessingException {

        String message = "相关信息检查失败";
        // 取值
        // JsonContext document = (JsonContext) JsonPath.parse(srcJson);

        // 分析结果
        QSTTY qstty = new QSTTY();
        // 校准标志
        qstty.setJzbz(jzbz);
        /** 检测数据主键(标识码)" **/
        qstty.setJcsjzj(jcsjzj);

        /** topic名称 **/
        qstty.setTopicmc(topic);

        /** 专项代码 **/
        qstty.setZxdm(zxdm);

        /** 城市 **/
        qstty.setCs(topicsscsdm);

        qstty.setZjgz("relevance");

        /** 字段分级 **/
        qstty.setZdfj("A");

        /** 质检规则分类 **/
        qstty.setZjgzfl("relevance");

        /** 质检时间 **/
        qstty.setZjsj(zjsj);

        /** 操作时间 **/
        qstty.setOperate_time(JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));

        /** 抽取时间 **/
        qstty.setExtract_time(extractTime);

        /** 问题字段名称 **/
        qstty.setWtzdmc("");

        /** 错误code **/
        qstty.setErrorCode("relevance");

        /** 错误消息 **/
        qstty.setErrorMessage(message);
        qstty.setCity_partition_id(city_partition_id);
        qstty.setCity_offset_id(city_offset_id);

        qstty.setCity_ts(city_ts);

        qstty.setClayer_partition(clayer_partition);
        qstty.setClayer_offset(clayer_offset);
        if (clayer_ts == null) {
            qstty.setClayer_ts("");
        } else {
            qstty.setClayer_ts(formatDate(clayer_ts));
        }

        String json = objectMapper.writeValueAsString(qstty);
        System.out.println("####################校验错误输出#####################");
        System.out.println(json);
        System.out.println("#########################################");
        return json;
    }


    /**
     * 质量分析
     *
     * @param jcsjzj
     * @param topicsscsdm
     * @param extractTime
     * @param topic
     * @param zjsj
     * @param srcDocument
     * @param jzzlDocument
     * @param validationMessage
     * @return
     */
    private String analysis(String jzbz, String zxdm, String jcsjzj, String topicsscsdm, String extractTime,
            String topic, String zjsj, JsonContext srcDocument, JsonContext jzzlDocument,
            ValidationMessage validationMessage, int city_partition_id, long city_offset_id, String city_ts,
            int clayer_partition, long clayer_offset, Long clayer_ts) throws JsonProcessingException {
        // json path
        String path = validationMessage.getPath();
        String code = validationMessage.getCode();
        String message = validationMessage.getMessage();
        if (path.equals("$")) {
            // 分割
            String[] msgs = message.split(":");
            // TODO 检查
            path = msgs[0];
            message = new String(validationMessage.getMessage().getBytes(StandardCharsets.ISO_8859_1),
                    StandardCharsets.UTF_8);
        }
        // 取值
        // JsonContext document = (JsonContext) JsonPath.parse(srcJson);

        // 分析结果
        QSTTY qstty = new QSTTY();
        // 校准标志
        qstty.setJzbz(jzbz);
        /** 检测数据主键(标识码)" **/
        qstty.setJcsjzj(jcsjzj);

        /** topic名称 **/
        qstty.setTopicmc(topic);

        /** 专项代码 **/
        qstty.setZxdm(zxdm);

        /** 城市 **/
        qstty.setCs(topicsscsdm);
        // 获取原始值
        try {
            Object value = srcDocument.read(path);
            /** 问题字段原始值 **/
            qstty.setWtzdysz(String.valueOf(value));
        } catch (Exception ex) {
            qstty.setWtzdysz(null);
        }

        /** 质检规则 **/
        // 根据code获得质检规则
        /**
         * String zjgzmsg = StreamProcess.properties.getProperty(code);
         * String zjgz ="";
         * String zjgzfl ="";
         * if(zjgzmsg!=null){
         * String[] msgs =zjgzmsg.split("|");
         * if(msgs.length==2){
         * zjgz = msgs[0];
         * zjgzfl= msgs[1];
         * }else{
         * zjgz = zjgzmsg;
         * }
         * }
         **/
        String zjgz = validationMessage.getType();
        String zjgzfl = message;
        qstty.setZjgz(zjgz);

        /** 字段分级 **/
        String zdfiPath = path + ".zdfj";
        try {
            qstty.setZdfj(jzzlDocument.read(zdfiPath, String.class));
        } catch (Exception e) {
            e.printStackTrace();
        }

        /** 质检规则分类 **/
        qstty.setZjgzfl(zjgzfl);

        /** 质检时间 **/
        qstty.setZjsj(zjsj);

        /** 操作时间 **/
        qstty.setOperate_time(JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));

        /** 抽取时间 **/
        qstty.setExtract_time(extractTime);

        /** 问题字段名称 **/
        qstty.setWtzdmc(path.replace("$.", ""));

        /** 错误code **/
        qstty.setErrorCode(code);

        /** 错误消息 **/
        qstty.setErrorMessage(message);
        qstty.setCity_partition_id(city_partition_id);
        qstty.setCity_offset_id(city_offset_id);

        qstty.setCity_ts(city_ts);

        qstty.setClayer_partition(clayer_partition);
        qstty.setClayer_offset(clayer_offset);
        if (clayer_ts == null) {
            qstty.setClayer_ts("");
        } else {
            qstty.setClayer_ts(formatDate(clayer_ts));
        }

        String json = objectMapper.writeValueAsString(qstty);
        System.out.println("####################校验错误输出#####################");
        System.out.println(json);
        System.out.println("#########################################");
        return json;
    }

    private List<String> getZdfjs(JsonContext jzzlDocument, Set<ValidationMessage> validates) {
        List<String> zdfjs = new ArrayList<String>();
        Iterator<ValidationMessage> iterator = validates.iterator();
        while (iterator.hasNext()) {
            ValidationMessage next = iterator.next();
            // json path
            String path = next.getPath();
            String message = new String(next.getMessage().getBytes(StandardCharsets.ISO_8859_1),
                    StandardCharsets.UTF_8);
            if (path.equals("$")) {
                // 分割
                String[] msgs = message.split(":");
                path = msgs[0];
            }
            String zdfiPath = path + ".zdfj";
            try {
                String zdfj = jzzlDocument.read(zdfiPath, String.class);
                zdfjs.add(zdfj);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return zdfjs;
    }

    /**
     * 解析及系统错误分析
     *
     * @param topic
     * @param cs
     * @param srcData
     * @return
     */
    private String analysisExecption(String topic, String cs, String srcData, String code, String message,
            int city_partition_id, long city_offset_id, String city_ts, int clayer_partition, long clayer_offset,
            Long clayer_ts) throws JsonProcessingException {
        ECTTY ectty = new ECTTY();
        /** topic名称 **/
        ectty.setTopicmc(topic);

        /** 城市 **/
        ectty.setCs(cs);

        /** 原始数据 **/
        ectty.setScrData(srcData);

        /** 错误代码 **/
        ectty.setErrorCode(code);
        /** 错误信息 **/
        ectty.setErrorMessage(message);
        ectty.setCity_partition_id(city_partition_id);
        ectty.setCity_offset_id(city_offset_id);
        ectty.setCity_ts(city_ts);

        ectty.setClayer_partition(clayer_partition);
        ectty.setClayer_offset(clayer_offset);
        if (clayer_ts == null) {
            ectty.setClayer_ts("");
        } else {
            ectty.setClayer_ts(formatDate(clayer_ts));
        }

        ectty.setOperate_time(JsonSchemaUtil.getNowTime("yyyy-MM-dd HH:mm:ss"));
        String json = objectMapper.writeValueAsString(ectty);
        return json;
    }

    private String formatDate(long s) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        // 设置时区为东八区（北京）
        TimeZone timeZone = TimeZone.getTimeZone("GMT+8");
        sdf.setTimeZone(timeZone);

        // 获取当前时间
        Date date = new Date(s);

        // 将日期按照指定的格式转换成字符串
        return sdf.format(date);
    }

}
